package com.gitee.checkpoint

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
import org.apache.spark.storage.StorageLevel

object CheckpointLearn {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession.builder().appName("cp").master("local[*]").getOrCreate()
    val sc: SparkContext = spark.sparkContext
    sc.setLogLevel("WARN")
    //设置checkpoint路径
    sc.setCheckpointDir("hdfs://node01:9000/spark/checkpoint")
    //计算数据
    val result: RDD[(String, Int)] = sc.textFile("data/words.txt").flatMap(_.split(" ")).map((_,1))reduceByKey(_+_)
    //对检查点保存
    result.checkpoint()
    result.foreach(println)//只有在操作Action算子时才会触发checkpoint
    //后续如果要使用就从checkpoint中读取

    /*
    * 我们在试试持久化
    * 分两种,一种缓存到磁盘,一种缓存到内存,但是都没有checkpoint好,因为它可以缓存到hdfs(容错)
    * Persist 和 Cache会在RDD结束后被删除,checkpoint会一直保存,不会被删除*/
    result.persist(StorageLevel.MEMORY_AND_DISK) //因为是零食的,所以我也不知道存储在了那里
    result.unpersist(true) //当RDD结束我们手动把它清除
  }
}
